#include "NetMgr.h"
#include <forward_list>
#include <sstream>
//#include "llLog.h"

NetMgr NetMgr::m_oInstance;


class EventN
{
public:
	int32_t Wait()
	{
		std::unique_lock<decltype(m_MutexCondition)> lock{ m_MutexCondition };
		m_Condition.wait(lock, [&]()->bool { return ret != -1; }); // suspend and wait ...	
		return ret;
	}
	void Set()
	{
		std::unique_lock<decltype(m_MutexCondition)> lock(m_MutexCondition);
		ret = 0;
		m_Condition.notify_one();
	}
private:
	std::mutex m_MutexCondition;
	std::condition_variable m_Condition;
	int32_t ret = -1;
};

WorkerThreadN::WorkerThreadN()
{
}
WorkerThreadN::~WorkerThreadN()
{
	StopRun();
}

int32_t WorkerThreadN::UpdateChannel(ChannelPtr channel)
{

	//printf("### EventLoopN WorkerThreadN::UpdateChannel Start channel:%d ThreadID:%s this:%p \n", channel->GetSocket(), GetThreadId().c_str(), this);
	//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "### WorkerThreadN::UpdateChannel Start channel:%d ThreadID:%s this:%p ", channel->GetSocket(), GetThreadId().c_str(), this);
	int32_t ret = -1;
	InvokTask([this, &channel, &ret]()
		{
			//printf("*** EventLoopN WorkerThreadN::UpdateChannel InvokTask Start  channel:%d ThreadID:%s this:%p \n", channel->GetSocket(), GetThreadId().c_str(), this);
			//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "*** WorkerThreadN::UpdateChannel InvokTask Start  channel:%d ThreadID:%s this:%p ", channel->GetSocket(), GetThreadId().c_str(), this);
			socket_t socket = channel->GetSocket();

			std::unordered_map<socket_t, ChannelPtr>::iterator iter = channels_.find(socket);
			if (iter != channels_.end()) {
				if (channel->IsNoneEvent()) {

					channels_.erase(socket);
					ret = 1;
				}
				else {
					// 更新读写事件 暂时不考虑
					//iter->second->
					ret = 2;
				}
			}
			else {
				if (!channel->IsNoneEvent()) {
					channels_.emplace(socket, channel);
					ret = 0;
				}
				else
				{
					ret = 3;
				}
				//printf("channel :%d \n", channel->GetSocket());
			}

			//printf("*** EventLoopN WorkerThreadN::UpdateChannel InvokTask End  channel:%d ThreadID:%s ret:%d this:%p \n", channel->GetSocket(), GetThreadId().c_str(), ret, this);
			//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "*** WorkerThreadN::UpdateChannel InvokTask End  channel:%d ThreadID:%s ret:%d this:%p ", channel->GetSocket(), GetThreadId().c_str(), ret, this);
		}
	);
	//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "### WorkerThreadN::UpdateChannel End channel:%d ThreadID:%s ret:%d this:%p ", channel->GetSocket(), GetThreadId().c_str(), ret, this);
	//printf("### EventLoopN WorkerThreadN::UpdateChannel End channel:%d ThreadID:%s ret:%d this:%p \n", channel->GetSocket(), GetThreadId().c_str(), ret, this);
	return ret;
}

int32_t WorkerThreadN::RemoveChannel(ChannelPtr& channel)
{
	//printf("### EventLoopN WorkerThreadN::RemoveChannel Start channel:%d ThreadID:%s this:%p \n", channel->GetSocket(), GetThreadId().c_str(), this);
	//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "### WorkerThreadN::RemoveChannel Start channel:%d ThreadID:%s this:%p ", channel->GetSocket(), GetThreadId().c_str(), this);
	int ret = -1;
	InvokTask([&channel, &ret, this]()
		{
			//printf("*** EventLoopN WorkerThreadN::RemoveChannel InvokTask Start  channel:%d ThreadID:%s this:%p \n", channel->GetSocket(), GetThreadId().c_str(), this);
			//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "*** WorkerThreadN::RemoveChannel InvokTask Start  channel:%d ThreadID:%s this:%p ", channel->GetSocket(), GetThreadId().c_str(), this);

			socket_t fd = channel->GetSocket();

			if (channels_.find(fd) != channels_.end()) {
				channels_.erase(fd);
				ret = 0;
			}
			else
			{
				ret = 1;
			}
			//printf("*** EventLoopN WorkerThreadN::RemoveChannel InvokTask End channel:%d ThreadID:%s ret:%d this:%p \n", channel->GetSocket(), GetThreadId().c_str(), ret, this);
			//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "*** WorkerThreadN::RemoveChannel InvokTask End channel:%d ThreadID:%s ret:%d this:%p ", channel->GetSocket(), GetThreadId().c_str(), ret, this);
		}
	);
	//printf("### EventLoopN WorkerThreadN::RemoveChannel End channel:%d ThreadID:%s ret:%d this:%p \n", channel->GetSocket(), GetThreadId().c_str(), ret, this);
	//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "### WorkerThreadN::RemoveChannel End channel:%d ThreadID:%s ret:%d this:%p ", channel->GetSocket(), GetThreadId().c_str(), ret, this);
	return ret;
}

void WorkerThreadN::PostTask(Task task)
{
	// 调用线程和将要打入的线程是同一个
	if (_loopThreadId == std::this_thread::get_id())
	{
		// 原地执行就可以
		task();
		return;
	}

	{
		std::unique_lock<std::mutex> lock(_mutex);
		_queue.push_back([this, &task]() {

			task();

			});
	}
	char event = 1;
	m_oPipe.Write(&event, 1);
}


void WorkerThreadN::InvokTask(Task task)
{
	// 调用线程和将要打入的线程是同一个
	if (_loopThreadId == std::this_thread::get_id())
	{
		// 原地执行就可以
		task();
		return;
	}

	EventN tEvent;
	{
		std::unique_lock<std::mutex> lock(_mutex);
		_queue.push_back([task, &tEvent]() {
			task();
			tEvent.Set();
			});
	}
	char event = 1;
	m_oPipe.Write(&event, 1);
	//printf("Wait before \n");
	tEvent.Wait();
}

std::string WorkerThreadN::GetThreadId()
{
	std::ostringstream oss;
	oss << _loopThreadId;
	return oss.str();
}
int64_t WorkerThreadN::getCurrentTicket()
{
	//std::chrono::steady_clock::time_point t1 = std::chrono::steady_clock::now();
	//std::chrono::steady_clock::time_point t11;
	//int64_t duration = std::chrono::duration_cast<std::chrono::microseconds>(t1 - t11).count();
	//int64_t duration = std::chrono::duration_cast<std::chrono::milliseconds>(t1 - t11).count();
	return  std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
}


int64_t WorkerThreadN::getMinDelay() {
	auto it = m_delayTask.begin();
	if (it == m_delayTask.end()) {
		return 10;
	}
	auto now = getCurrentTicket();
	if (it->first > now) {
		//所有任务尚未到期
		////printf("it->first:%llu now:%lld dif:%lld \n", it->first, now, it->first - now);
		return it->first - now;
	}
	//执行已到期的任务并刷新休眠延时
	return flushDelayTask(now);
}

int64_t WorkerThreadN::flushDelayTask(uint64_t now_time) {

	////printf("############# sleep :%llu \n", now_time);
	decltype(m_delayTask) taskCopy;
	taskCopy.swap(m_delayTask);
	uint64_t start = getCurrentTicket();
	for (auto it = taskCopy.begin(); it != taskCopy.end() && it->first <= now_time; it = taskCopy.erase(it)) {
		//已到期的任务
		//auto next_delay = (*(it->second))();
		std::shared_ptr<TimerN>  pTimer = it->second;
		if (pTimer->m_cb)
		{
			bool ret = pTimer->m_cb();
			if (ret)
			{
				m_delayTask.emplace(pTimer->m_iDelay + now_time, std::move(it->second));
			}
			else
			{
				//MEDIA_TRACE_LOG("EventLoop", LL_INFO, "TaskScheduler::flushDelayTask this:%p m_iTimerId:%lld ret:%d", this, pTimer->m_iTimerId, ret);
			}
		}
	}
	uint64_t end = getCurrentTicket();
	int64_t useTime = end - start;
	if (useTime > 11)
	{
		////printf("timers_poll::flushDelayTask useTimer = %d \n", useTime);
	}

	taskCopy.insert(m_delayTask.begin(), m_delayTask.end());
	taskCopy.swap(m_delayTask);

	auto it = m_delayTask.begin();
	if (it == m_delayTask.end()) {
		//没有剩余的定时器了
		return 0;
	}
	//最近一个定时器的执行延时

	int64_t t_sleep = it->first - getCurrentTicket();
	if (t_sleep < -1)
	{
		//printf("timers_poll::flushDelayTask  t_sleep = %lld  \n", t_sleep);
	}
	return t_sleep > 0 ? t_sleep : 0;

}

int64_t WorkerThreadN::AddTimer(int64_t iDelay, std::function<bool()> fcb)
{

	//printf("### EventLoop TaskScheduler::AddTimer Start iDelay:%lld ThreadID:%s this:%p \n", iDelay, GetThreadId().c_str(), this);
	//MEDIA_TRACE_LOG("EventLoop", LL_INFO, "### TaskScheduler::AddTimer Start iDelay:%lld ThreadID:%s this:%p ", iDelay, GetThreadId().c_str(), this);
	EventN tEvent;
	auto t_timer = std::make_shared<TimerN>(iDelay, fcb);
	int64_t t_timer_id = 0;
	InvokTask([this, &t_timer, &t_timer_id]()
		{
			//printf("*** EventLoop TaskScheduler::AddTimer InvokTask Start   ThreadID:%s this:%p \n",  GetThreadId().c_str(), this);
			//MEDIA_TRACE_LOG("EventLoop", LL_INFO, "*** TaskScheduler::AddTimer InvokTask Start   ThreadID:%s this:%p ", GetThreadId().c_str(), this);

			m_iTimerId++;
			t_timer->m_iTimerId = m_iTimerId;
			t_timer_id = m_iTimerId;
			uint64_t time_line = getCurrentTicket() + t_timer->m_iDelay;
			m_delayTask.emplace(time_line, std::move(t_timer));

			//printf("*** EventLoop TaskScheduler::AddTimer InvokTask End ThreadID:%s this:%p \n",  GetThreadId().c_str(), this);
			//MEDIA_TRACE_LOG("EventLoop", LL_INFO, "*** TaskScheduler::AddTimer InvokTask End ThreadID:%s this:%p ", GetThreadId().c_str(), this);
		}
	);

	//printf("### EventLoop TaskScheduler::AddTimer End iDelay:%lld ThreadID:%s t_timer_id:%lld this:%p \n", iDelay, GetThreadId().c_str(), t_timer_id, this);
	//MEDIA_TRACE_LOG("EventLoop", LL_INFO, "### TaskScheduler::AddTimer End iDelay:%lld ThreadID:%s t_timer_id:%lld this:%p ", iDelay, GetThreadId().c_str(), t_timer_id, this);
	return t_timer_id;
}

int32_t WorkerThreadN::DelTimer(int64_t timerId)
{
	//printf("### EventLoop TaskScheduler::DelTimer Start timerId:%lld ThreadID:%s this:%p \n", timerId, GetThreadId().c_str(), this);
	//MEDIA_TRACE_LOG("EventLoop", LL_INFO, "### TaskScheduler::DelTimer Start timerId:%lld ThreadID:%s this:%p ", timerId, GetThreadId().c_str(), this);
	int ret = -1;
	InvokTask([this, timerId, &ret]()
		{
			//printf("*** EventLoop TaskScheduler::DelTimer Start InvokTask timerId:%lld ThreadID:%s this:%p \n", timerId, GetThreadId().c_str(), this);
			//MEDIA_TRACE_LOG("EventLoop", LL_INFO, "*** TaskScheduler::DelTimer Start InvokTask timerId:%lld ThreadID:%s this:%p ", timerId, GetThreadId().c_str(), this);
			std::multimap<uint64_t, std::shared_ptr<TimerN> >::iterator it = m_delayTask.begin();
			for (; it != m_delayTask.end();)
			{
				std::shared_ptr<TimerN> pTimer = it->second;
				if (pTimer->m_iTimerId == timerId)
				{
					it = m_delayTask.erase(it);
					ret = 0;
					break;
				}
				else
				{
					it++;
				}
			}

			//printf("*** EventLoop TaskScheduler::DelTimer End InvokTask timerId:%lld ThreadID:%s ret:%d this:%p \n", timerId, GetThreadId().c_str(), ret, this);
			//MEDIA_TRACE_LOG("EventLoop", LL_INFO, "*** TaskScheduler::DelTimer End InvokTask timerId:%lld ThreadID:%s ret:%d this:%p ", timerId, GetThreadId().c_str(), ret, this);

		});

	//printf("### EventLoop TaskScheduler::DelTimer End timerId:%lld ThreadID:%s ret:%d this:%p \n", timerId, GetThreadId().c_str(), ret, this);
	//MEDIA_TRACE_LOG("EventLoop", LL_INFO, "### TaskScheduler::DelTimer End timerId:%lld ThreadID:%s ret:%d this:%p ", timerId, GetThreadId().c_str(), ret, this);
	return ret;
}

int32_t WorkerThreadN::StartRun()
{

	int ret = m_oPipe.Create();
	if (0 != ret)
	{
		return -2;
	}

	m_loopThread = new std::thread(&WorkerThreadN::RunLoop, this);
	while (!m_bStart)
	{
		std::this_thread::sleep_for(std::chrono::milliseconds(1));
	}


	if (NULL == m_loopThread)
	{
		return -1;
	}
	
	return 0;

}
void WorkerThreadN::StopRun()
{
	m_bExitThread = true;

	// 驱动线程快速退出
	char event = 1;
	m_oPipe.Write(&event, 1);

	if (m_loopThread)
	{
		m_loopThread->join();
		delete m_loopThread;
		m_loopThread = 0;
	}

	m_oPipe.Close();
}



void WorkerThreadN::RunLoop()
{
	_loopThreadId = std::this_thread::get_id();
	m_bStart = true;
	//_condition.notify_one();

	fd_set fd_read;
	fd_set fd_write;
	fd_set fd_exp;


	while (!m_bExitThread)
	{
		int max_fd = 0;

		FD_ZERO(&fd_read);
		FD_ZERO(&fd_write);
		FD_ZERO(&fd_exp);


		FD_SET(m_oPipe.Read(), &fd_read);


		max_fd = m_oPipe.Read();

		for (auto iter : channels_) {
			int events = iter.second->GetEvents();
			socket_t fd = iter.second->GetSocket();

			if ((events & EVENT_IN)) {
				FD_SET(fd, &fd_read);
			}

			if ((events & EVENT_OUT)) {
				FD_SET(fd, &fd_write);
			}
			if (fd > max_fd)
			{
				max_fd = fd;
			}

		}
		struct timeval tv;
		int32_t minDelay = 20;
		tv.tv_sec = minDelay / 1000;
		tv.tv_usec = 1000 * (minDelay % 1000);
		int ret = select(max_fd + 1, &fd_read, &fd_write, NULL, &tv);

		if (ret > 0)
		{

			std::forward_list<std::pair<ChannelPtr, int>> event_list;
			 {

				for (auto iter : channels_) {
					int events = 0;
					socket_t socket = iter.second->GetSocket();

					if (FD_ISSET(socket, &fd_read)) {
						events |= EVENT_IN;
					}

					if (FD_ISSET(socket, &fd_write)) {
						events |= EVENT_OUT;
					}

					if (FD_ISSET(socket, &fd_exp)) {
						events |= (EVENT_HUP); // close
					}

					if (events != 0) {
						event_list.emplace_front(iter.second, events);
					}
				}
			}

			for (auto& iter : event_list) {
				iter.first->HandleEvent(iter.second);
			}
			
			if (FD_ISSET(m_oPipe.Read(), &fd_read))
			{
				char buf[1];
				m_oPipe.Read(buf, sizeof(buf));		
			}
		}

		// 处理增删任务
		std::list<Task> t_queue;
		{
			std::unique_lock<std::mutex> lock(_mutex);
			if (!_queue.empty())
			{
				t_queue.swap(_queue);

			}
		}

		// 处理任务
		if (!t_queue.empty())
		{
			for (std::list<Task>::iterator it = t_queue.begin(); it != t_queue.end(); it++)
			{
				Task task = *it;
				task();
			}
			t_queue.clear();
			//printf("处理管道任务\n");
		}
		
	
	}

}

EventLoopN::EventLoopN()
{

}

EventLoopN::~EventLoopN()
{
	Quit();
}

int32_t EventLoopN::Loop(uint32_t num_threads)
{
	std::lock_guard<std::mutex> locker(mutex_);
	num_threads_ = num_threads;
	//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "### EventLoopN::Loop Start num_threads:%d this:%p ", num_threads, this);
	if (!task_schedulers_.empty()) {
		//printf("### EventLoopN Have Looped !task_schedulers_.empty() this:%p task_schedulers_.size():%d \n", this, task_schedulers_.size());
		//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "### EventLoopN End Have Looped !task_schedulers_.empty() this:%p task_schedulers_.size():%d ", this, task_schedulers_.size());
		return task_schedulers_.size();
	}
	for (int32_t n = 0; n < num_threads_; n++)
	{
		int32_t ret = 0;
		//printf("*** EventLoopN::Loop StartRun before n:%d this:%p num_threads_:%d \n", n, this, num_threads_);
		//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "*** EventLoopN::Loop StartRun before n:%d this:%p ", n, this);
		std::shared_ptr<WorkerThreadN> task_scheduler_ptr(new WorkerThreadN());
		ret = task_scheduler_ptr->StartRun();

		//printf("*** EventLoopN::Loop StartRun end n:%d ret:%d task_scheduler_ptr:%p ThreadID:%s this:%p \n", n, ret, task_scheduler_ptr.get(), task_scheduler_ptr->GetThreadId().c_str(), this);
		//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "*** EventLoopN::Loop StartRun end n:%d ret:%d task_scheduler_ptr:%p ThreadID:%s this:%p ", n, ret, task_scheduler_ptr.get(), task_scheduler_ptr->GetThreadId().c_str(), this);
		if (0 == ret)
		{
			task_schedulers_.push_back(task_scheduler_ptr);
		}
	
	}
	//MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "### EventLoopN::Loop End task_schedulers_.size():%d this:%p ", task_schedulers_.size(), this);
	return task_schedulers_.size();
}

void EventLoopN::Quit()
{
	//printf("EventLoopN::Quit Start this:%p task_schedulers_.size():%d \n", this, task_schedulers_.size());
	////MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "### EventLoopN::Quit Start this:%p task_schedulers_.size():%d ", this, task_schedulers_.size());
	int i = 0;
	for (auto iter : task_schedulers_)
	{
		std::shared_ptr<WorkerThreadN> t_scheduler = iter;
		//printf("EventLoopN::StopRun before i:%d iter:%p ThreadID:%s this:%p \n", i, t_scheduler.get(), t_scheduler->GetThreadId().c_str(), this);
		////MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "*** EventLoopN::StopRun before i:%d iter:%p ThreadID:%s this:%p ", i, t_scheduler.get(), t_scheduler->GetThreadId().c_str(), this);
		iter->StopRun();
		//printf("EventLoopN::StopRun after i:%d iter:%p ThreadID:%s this:%p \n", i, t_scheduler.get(), t_scheduler->GetThreadId().c_str(), this);
		////MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "*** EventLoopN::StopRun after i:%d iter:%p ThreadID:%s this:%p ", i, t_scheduler.get(), t_scheduler->GetThreadId().c_str(), this);
		i++;
	}
	task_schedulers_.clear();
	//printf("EventLoopN::Quit End this:%p \n", this);
	////MEDIA_TRACE_LOG("EventLoopN", LL_INFO, "### EventLoopN::Quit End this:%p ", this);
}

std::shared_ptr<WorkerThreadN> EventLoopN::GetTaskScheduler()
{
	std::lock_guard<std::mutex> locker(mutex_);
	if (task_schedulers_.size() == 1) {
		return task_schedulers_.at(0);
	}
	else {
		auto task_scheduler = task_schedulers_.at(index_);
		index_++;
		if (index_ >= task_schedulers_.size()) {
			index_ = 0;
		}
		return task_scheduler;
	}

	return nullptr;
}

void EventLoopN::UpdateChannel(ChannelPtr channel)
{
	std::lock_guard<std::mutex> locker(mutex_);
	if (task_schedulers_.size() > 0) {
		task_schedulers_[0]->UpdateChannel(channel);
	}
}

void EventLoopN::RemoveChannel(ChannelPtr& channel)
{
	std::lock_guard<std::mutex> locker(mutex_);
	if (task_schedulers_.size() > 0) {
		task_schedulers_[0]->RemoveChannel(channel);
	}
}

NetMgr::NetMgr()
{
}

NetMgr* NetMgr::GetInstance()
{
	return &(NetMgr::m_oInstance);
	
}

int32_t NetMgr::Init(uint32_t num_threads)
{
	int32_t ret = m_oEventLoop.Loop(num_threads);
	return ret;
}

NetMgr::~NetMgr()
{
	m_oEventLoop.Quit();
}